package com.pw.study.flink.sql;

import com.pw.study.flink.entities.WaterSensor;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;

import static org.apache.flink.table.api.Expressions.$;

/**
 * @Author: linux_future
 * @since: 2022/3/11
 **/
public class $11SqlWindow_TVF {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        // 1. 先创建一个流
        DataStream<WaterSensor> stream = env
                .fromElements(
                        new WaterSensor("sensor_1", 1000L, 10),
                        new WaterSensor("sensor_1", 2000L, 20),
                        new WaterSensor("sensor_2", 3000L, 30),
                        new WaterSensor("sensor_1", 4001L, 40),
                        new WaterSensor("sensor_1", 5000L, 50),
                        new WaterSensor("sensor_2", 6000L, 60)
                )
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy
                                .<WaterSensor>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                                .withTimestampAssigner((ws, ts) -> ws.getTs())

                );

        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);

        Table table = tEnv.fromDataStream(stream, $("id"), $("ts").rowtime(), $("vc"));
        tEnv.createTemporaryView("sensor", table);

        tvf1(tEnv);
        tvf2(tEnv);
        tvf3(tEnv);
        tvf4(tEnv);
        tvf5(tEnv);
        tvf6(tEnv);
        tvf7(tEnv);
        tvf8(tEnv);


    }

    private static void tvf8(StreamTableEnvironment tEnv) {
    }

    private static void tvf7(StreamTableEnvironment tEnv) {
    }

    private static void tvf6(StreamTableEnvironment tEnv) {
        System.out.println("===============累积窗口6================");
        tEnv
                .sqlQuery("select" +
                        " 'a' id, window_start, window_end, " +
                        " sum(vc) vc_sum " +
                        "from table( tumble(table sensor, descriptor(ts), interval '5' second) )" +
                        "group by window_start, window_end " +
                        "union " +
                        "select" +
                        " id, window_start, window_end, " +
                        " sum(vc) vc_sum " +
                        "from table( tumble(table sensor, descriptor(ts), interval '5' second) )" +
                        "group by id, window_start, window_end " +
                        "")
                .execute()
                .print();
    }

    private static void tvf5(StreamTableEnvironment tEnv) {
        System.out.println("===============累积窗口================");
        tEnv
                .sqlQuery("select" +
                        " id, window_start, window_end, " +
                        " sum(vc) vc_sum " +
                        "from table( tumble(table sensor, descriptor(ts), interval '5' second) )" +
                        "group by window_start, window_end, grouping sets( (id), () )" +
                        "")
                .execute()
                .print();
    }

    private static void tvf4(StreamTableEnvironment tEnv) {
        System.out.println("tvf window 累积窗口....................");
        // 累积窗口
        tEnv
                .sqlQuery("select" +
                        " id, window_start, window_end, " +
                        " sum(vc) vc_sum " +
                        "from table( cumulate(table sensor, descriptor(ts), interval '2' second, interval '10' second) )" +
                        "group by id, window_start, window_end")
                .execute()
                .print();
    }

    private static void tvf3(StreamTableEnvironment tEnv) {
        System.out.println("window tumble1 ................");
        tEnv
                .sqlQuery("select" +
                        " id, window_start, window_end, " +
                        " sum(vc) vc_sum " +
                        "from table(tumble( table sensor, descriptor(ts), interval '5' second  ))" +
                        "group by id, window_start, window_end")
                .execute()
                .print();
    }

    private static void tvf2(StreamTableEnvironment tEnv) {
        System.out.println("window tumble2 ................");
        tEnv
                .sqlQuery("select" +
                        " id, window_start, window_end, " +
                        " sum(vc) vc_sum " +
                        "from table(tumble(DATA => table sensor,TIMECOL => descriptor(ts), SIZE =>interval '5' second  ))" +
                        "group by id, window_start, window_end")
                .execute()
                .print();
    }

    private static void tvf1(StreamTableEnvironment tEnv) {
        System.out.println("tvf...滑动窗口的长度必须是滑动步长的整数倍");
        tEnv
                .sqlQuery("select" +
                        " id, window_start, window_end, " +
                        " sum(vc) vc_sum " +
                        "from table(hop(table sensor, descriptor(ts), interval '2' second, interval '4' second  ))" +
                        "group by id, window_start, window_end")
                .execute()
                .print();
    }

}
